package org.apache.flink.endpoint;

import org.apache.flink.gateway.ResourceManagerGateway;
import org.apache.flink.gateway.TaskExecutorRpcGateway;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;

import java.util.Timer;
import java.util.TimerTask;

public class TaskExecutorRpcEndpoint extends RpcEndpoint implements TaskExecutorRpcGateway {

    private Timer heatbeatTimer;

    public TaskExecutorRpcEndpoint(RpcService rpcService, String endpointId) {
        super(rpcService, endpointId);
        heatbeatTimer = new Timer();
    }

    @Override
    protected void onStart() throws Exception {
        System.out.println("TaskExecutorRpcEndpoint onStart");

        String address = "akka.tcp://flink@127.0.0.1:9002/user/rpc/rm";
        ResourceManagerGateway resourceManagerGateway = getRpcService().connect(address, ResourceManagerGateway.class).get();
        resourceManagerGateway.registerFromTaskExecutor(getAddress());

        heatbeatTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                resourceManagerGateway.heatBeatFromJobmanager(getAddress());
            }
        }, 2000, 2000);
    }
}
